package com.hadoop.mapreduce.Partitioner2;

import entity.UserEntity;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/*
 * Reducer需要定义四个输出、输出类型泛型：
 * 四个泛型类型分别代表：
 * KeyIn        Reducer的输入数据的Key，这里是Text UserEntity
 * ValueIn      Reducer的输入数据的Value，这里是序列化对象UserEntity
 * KeyOut       Reducer的输出数据的Key，这里是序列化对象UserEntity
 * ValueOut     Reducer的输出数据的Value，NullWritable
 *
 * UserEntity 的compareTo 要实现，
 * 一个key调一次这里是把数据当作key在用，所以会快速调多次，
 * 所以可能会出现 Bad file descriptor（文件还在用又要打开） 这个错误，但不影响，
 *
 */
public class UserReducer extends Reducer<UserEntity,NullWritable,UserEntity,NullWritable> {


    @Override
    protected void reduce(UserEntity userEntity, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
//        for (UserEntity userEntity : values) {
//            // 年收入 = 月收入 * 12  四舍五入
//            String yearIncome = String.format("%.2f", userEntity.getMonthIncome() * 12);
//            userEntity.setYearIncome(Double.parseDouble(yearIncome));
//            System.out.println("****************UserReducer "+Thread.currentThread().getId()+"**********************");
//            System.out.println("UserReducer - reduce ="+userEntity.toString());
//            context.write(userEntity, NullWritable.get());
//        }
        int size = 0;
        for (NullWritable n:values) {
            size++;
        }
        // 年收入 = 月收入 * 12  四舍五入
        String yearIncome = String.format("%.2f", userEntity.getMonthIncome() * 12);
        userEntity.setYearIncome(Double.parseDouble(yearIncome));
        System.out.println("*************Thread id = "+Thread.currentThread().getId()+
                "this.add="+this.hashCode()+";size="+size);
        context.write(userEntity, NullWritable.get());
    }
}
